//! 网络故障模拟测试

use nacos_cluster::{ClusterConfig, ClusterManager, ClusterNodeInfo, NodeState};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

/// 模拟网络延迟
#[tokio::test]
async fn test_network_latency_simulation() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "latency-test".to_string(),
        heartbeat_timeout: 3, // 3秒超时
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    // 创建节点
    let nodes = vec![
        ClusterNodeInfo::new("127.0.0.1", 8848, 1),
        ClusterNodeInfo::new("127.0.0.1", 8849, 1),
        ClusterNodeInfo::new("127.0.0.1", 8850, 1),
    ];
    
    manager.initialize(nodes[0].clone()).await.unwrap();
    manager.add_node(nodes[1].clone()).await.unwrap();
    manager.add_node(nodes[2].clone()).await.unwrap();
    
    // 启动心跳检查
    let manager_clone = manager.clone();
    manager.start_heartbeat_check().await;
    
    // 模拟网络延迟，不更新心跳
    sleep(Duration::from_secs(4)).await;
    
    // 所有节点应该被标记为故障
    let stats = manager.get_cluster_stats();
    assert_eq!(stats.healthy_nodes, 0);
    assert_eq!(stats.cluster_state, ClusterState::Stopped);
    
    // 恢复网络连接
    for node in &nodes {
        manager.update_node_heartbeat(&node.node_id).await.unwrap();
    }
    
    // 所有节点应该恢复
    let stats = manager.get_cluster_stats();
    assert_eq!(stats.healthy_nodes, 3);
    assert_eq!(stats.cluster_state, ClusterState::Running);
}

/// 模拟网络分区
#[tokio::test]
async fn test_network_partition_scenarios() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "partition-scenarios".to_string(),
        heartbeat_timeout: 1,
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    // 创建5个节点的集群
    let nodes = vec![
        ClusterNodeInfo::new("127.0.0.1", 8848, 1),
        ClusterNodeInfo::new("127.0.0.1", 8849, 1),
        ClusterNodeInfo::new("127.0.0.1", 8850, 1),
        ClusterNodeInfo::new("127.0.0.1", 8851, 1),
        ClusterNodeInfo::new("127.0.0.1", 8852, 1),
    ];
    
    manager.initialize(nodes[0].clone()).await.unwrap();
    for node in &nodes[1..] {
        manager.add_node(node.clone()).await.unwrap();
    }
    
    // 设置初始领导者
    manager.set_leader(&nodes[0].node_id).await.unwrap();
    
    // 场景1：少数节点故障（2/5）
    manager.update_node_state(&nodes[3].node_id, NodeState::Failed).await.unwrap();
    manager.update_node_state(&nodes[4].node_id, NodeState::Failed).await.unwrap();
    
    // 集群应该仍然运行（3/5 = 60% > 50%）
    assert_eq!(manager.get_cluster_state(), ClusterState::Running);
    
    // 场景2：多数节点故障（3/5）
    manager.update_node_state(&nodes[2].node_id, NodeState::Failed).await.unwrap();
    
    // 集群应该进入降级模式（2/5 = 40% < 50%）
    assert_eq!(manager.get_cluster_state(), ClusterState::Degraded);
    
    // 场景3：恢复一个节点
    manager.update_node_state(&nodes[2].node_id, NodeState::Running).await.unwrap();
    
    // 集群应该恢复运行（3/5 = 60% > 50%）
    assert_eq!(manager.get_cluster_state(), ClusterState::Running);
}

/// 模拟脑裂场景
#[tokio::test]
async fn test_split_brain_scenario() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "split-brain".to_string(),
        heartbeat_timeout: 1,
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    // 创建偶数个节点的集群（容易脑裂）
    let nodes = vec![
        ClusterNodeInfo::new("127.0.0.1", 8848, 1),
        ClusterNodeInfo::new("127.0.0.1", 8849, 1),
        ClusterNodeInfo::new("127.0.0.1", 8850, 1),
        ClusterNodeInfo::new("127.0.0.1", 8851, 1),
    ];
    
    manager.initialize(nodes[0].clone()).await.unwrap();
    for node in &nodes[1..] {
        manager.add_node(node.clone()).await.unwrap();
    }
    
    // 设置初始领导者
    manager.set_leader(&nodes[0].node_id).await.unwrap();
    
    // 网络分区：分成两个2节点的小集群
    manager.update_node_state(&nodes[2].node_id, NodeState::Failed).await.unwrap();
    manager.update_node_state(&nodes[3].node_id, NodeState::Failed).await.unwrap();
    
    // 集群应该进入降级模式（2/4 = 50%）
    assert_eq!(manager.get_cluster_state(), ClusterState::Degraded);
    
    // 领导者应该仍然是原来的领导者
    assert_eq!(manager.get_leader().unwrap().node_id, nodes[0].node_id);
}

/// 模拟节点频繁上下线
#[tokio::test]
async fn test_flapping_nodes() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "flapping-test".to_string(),
        heartbeat_timeout: 1,
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    let node = ClusterNodeInfo::new("127.0.0.1", 8848, 1);
    manager.initialize(node.clone()).await.unwrap();
    
    // 模拟节点频繁上下线
    for _ in 0..5 {
        manager.update_node_state(&node.node_id, NodeState::Failed).await.unwrap();
        sleep(Duration::from_millis(100)).await;
        manager.update_node_state(&node.node_id, NodeState::Running).await.unwrap();
        sleep(Duration::from_millis(100)).await;
    }
    
    // 最终节点应该是健康的
    let stats = manager.get_cluster_stats();
    assert_eq!(stats.healthy_nodes, 1);
    assert_eq!(stats.cluster_state, ClusterState::Running);
}

/// 模拟大规模节点故障恢复
#[tokio::test]
async fn test_massive_failure_recovery() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "massive-recovery".to_string(),
        heartbeat_timeout: 1,
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    // 创建10个节点的集群
    let mut nodes = vec![];
    for i in 0..10 {
        let node = ClusterNodeInfo::new("127.0.0.1", 8848 + i, 1);
        nodes.push(node);
    }
    
    manager.initialize(nodes[0].clone()).await.unwrap();
    for node in &nodes[1..] {
        manager.add_node(node.clone()).await.unwrap();
    }
    
    // 设置领导者
    manager.set_leader(&nodes[0].node_id).await.unwrap();
    
    // 模拟大规模故障（9/10节点故障）
    for node in &nodes[1..] {
        manager.update_node_state(&node.node_id, NodeState::Failed).await.unwrap();
    }
    
    // 集群应该进入降级模式
    assert_eq!(manager.get_cluster_state(), ClusterState::Degraded);
    
    // 恢复所有节点
    for node in &nodes[1..] {
        manager.update_node_state(&node.node_id, NodeState::Running).await.unwrap();
    }
    
    // 集群应该恢复运行
    assert_eq!(manager.get_cluster_state(), ClusterState::Running);
    assert_eq!(manager.get_healthy_nodes().len(), 10);
}

/// 测试网络延迟下的领导者选举
#[tokio::test]
async fn test_election_with_latency() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    
    let config = ClusterConfig {
        cluster_name: "latency-election".to_string(),
        heartbeat_timeout: 3,
        ..Default::default()
    };
    
    let manager = Arc::new(ClusterManager::new(config));
    manager.set_event_sender(tx);
    
    // 创建不同权重的节点
    let nodes = vec![
        ClusterNodeInfo::new("127.0.0.1", 8848, 1),
        ClusterNodeInfo::new("127.0.0.1", 8849, 3), // 高权重
        ClusterNodeInfo::new("127.0.0.1", 8850, 2),
    ];
    
    manager.initialize(nodes[0].clone()).await.unwrap();
    for node in &nodes[1..] {
        manager.add_node(node.clone()).await.unwrap();
    }
    
    // 初始领导者
    manager.set_leader(&nodes[0].node_id).await.unwrap();
    
    // 模拟领导者故障
    manager.update_node_state(&nodes[0].node_id, NodeState::Failed).await.unwrap();
    
    // 新的领导者应该是权重最高的节点
    let new_leader = manager.get_leader().unwrap();
    assert_eq!(new_leader.node_id, nodes[1].node_id);
    assert_eq!(new_leader.weight, 3);
}